package cn.migu.music.hangupsms.kafka;

import java.util.Optional;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import cn.migu.music.common.utils.LogUtils;
import cn.migu.music.hangupsms.common.config.HangUpSmsConfig;
import cn.migu.music.hangupsms.common.entity.BillInfo;
import cn.migu.music.hangupsms.service.BaseService;
import lombok.extern.slf4j.Slf4j;

/**
 * @version 1.0 created by duanhao@migu.cn on 2022/3/7 9:55 上午
 */
@Component
@Slf4j
public class HangUpListener {
    @Autowired
    private HangUpSmsConfig hangUpSmsConfig;
    @Autowired
    private BaseService     baseService;

    /**
     * 从kafka消费话单数e
     *
     * @param record
     */
    @KafkaListener(topics = { "${kafka.call-list.topic}" })
    public void handleKafkaMessage(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            String content = (String) kafkaMessage.get();

            if (hangUpSmsConfig.isNeedHandleKafka()) {
                try {
                    BillInfo billInfo = baseService.getBillInfo(content);
                    if (baseService.contentPreProcess(billInfo)) {
                        baseService.contentProcess(billInfo, true);
                    }
                } catch (Exception businessException) {
                    LogUtils.info(businessException, log, "线程池队列已满");
                }
            } else {
                if (hangUpSmsConfig.isNeedLog()){
                    LogUtils.info(log, "接收到kafka消息:开关:%s,话单内容:%s", hangUpSmsConfig.isNeedHandleKafka(), content);
                }
            }
        }
    }

}
